package com.we.risk.phoneRecord.stage2.dimuser;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.we.flink.userlabelplatform.table.TableUserAccount;
import com.we.risk.phoneRecord.common.Stage2GlobalSetting;
import com.weicai.caesar.CaesarUtil;
import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.util.HashMap;
import java.util.List;

@Deprecated
public class TagUserAccountMem {
    public long uid;
    public String mobile;
    public String user_key;
    public int updated_at;

    public static Logger LOG = LoggerFactory.getLogger(TagUserAccountMem.class);
    public static final int MOBILELENGTH = 7;
    public static final String INPUT_KEY_BY = "uid";
    public static final String SPLITWORD = "#";

    public TagUserAccountMem() {}

    public TagUserAccountMem(long uid, String mobile, String user_key, int updated_at) {
        this.uid = uid;
        this.mobile = mobile;
        this.user_key = user_key;
        this.updated_at = updated_at;
    }

    public static TagUserAccountMem init(long uid) {
        return new TagUserAccountMem(uid, "", "", 0);
    }

    public static SingleOutputStreamOperator<String> getUserAccountProcessedStream(
            StreamTableEnvironment tableEnv, StreamExecutionEnvironment env) {
        Table tableRes = tableEnv.sqlQuery(TableUserAccount.QUERYSQL);
        DataStream<Tuple2<Boolean, TableUserAccount>> inputStream =
                tableEnv.toRetractStream(tableRes, TableUserAccount.class);

        env.registerCachedFile(
                Stage2GlobalSetting.MOBAREADATA, Stage2GlobalSetting.MOBAREAFILENAME);

        SingleOutputStreamOperator<String> out =
                inputStream
                        .filter(
                                new FilterFunction<Tuple2<Boolean, TableUserAccount>>() {
                                    @Override
                                    public boolean filter(Tuple2<Boolean, TableUserAccount> value)
                                            throws Exception {
                                        return value.f0;
                                    }
                                })
                        .uid("TagUserAccount-filter")
                        .keyBy(new TagUserAccountKeySelectorFunc())
                        .flatMap(new TagUserAccountFlatMapFunc());

        return out;
    }

    private static class TagUserAccountKeySelectorFunc
            implements KeySelector<Tuple2<Boolean, TableUserAccount>, Long> {
        @Override
        public Long getKey(Tuple2<Boolean, TableUserAccount> value) throws Exception {
            return value.f1.getUid();
        }
    }

    private static class TagUserAccountFlatMapFunc
            extends RichFlatMapFunction<Tuple2<Boolean, TableUserAccount>, String> {
        ValueState<TagUserAccountMem> userAccountVState;
        HashMap<String, String> mobAreaMap;

        @Override
        public void open(Configuration parameters) throws Exception {
            userAccountVState =
                    getRuntimeContext()
                            .getState(
                                    new ValueStateDescriptor<TagUserAccountMem>(
                                            "userAccountVState", TagUserAccountMem.class));
            initMobileAreaMap();
        }

        @Override
        public void flatMap(Tuple2<Boolean, TableUserAccount> value, Collector<String> out)
                throws Exception {
            try {
                long uid = value.f1.getUid();
                String user_key = value.f1.getUser_key();
                String mobile = value.f1.getRegister_mobile();
                /** cacser 加密 */
                if (CaesarUtil.isEncrypted(mobile)) {
                    mobile = CaesarUtil.decode(mobile);
                }
                if (mobile.length() >= MOBILELENGTH) {
                    mobile = mobile.substring(0, MOBILELENGTH);
                }
                int updated_at = value.f1.getUpdated_at();

                TagUserAccountMem tmpState = userAccountVState.value();
                TagUserAccountMem curState =
                        tmpState == null ? TagUserAccountMem.init(uid) : tmpState;

                if (curState.getUpdated_at() < updated_at) {
                    if (!curState.getMobile().equals(mobile)
                            || !curState.getUser_key().equals(user_key)) {
                        curState.mobile = mobile;
                        curState.user_key = user_key;
                        curState.updated_at = updated_at;

                        String res = mobAreaMap.get(mobile);
                        if (res != null) {
                            String[] splits = res.split(SPLITWORD);
                            if (splits.length == 2) {
                                JSONObject outJson = new JSONObject();
                                outJson.put("uid", uid);
                                outJson.put("user_key", user_key);
                                outJson.put("mobile", mobile);
                                outJson.put(
                                        "mob_prov", splits[0].equals("null") ? null : splits[0]);
                                outJson.put(
                                        "mob_city", splits[1].equals("null") ? null : splits[1]);
                                out.collect(outJson.toString(SerializerFeature.WriteMapNullValue));
                            } else {
                                LOG.error("TagUserAccountFlatMapFunc splits error res: " + res);
                            }
                        } else {
                            LOG.warn("UserJMobAreaPrcFun ProcData mob : " + mobile);
                        }
                        userAccountVState.update(curState);
                    }
                }
            } catch (Exception e) {
                LOG.error("TagUserAccount FlatMap Exception: " + e.toString());
                LOG.error("TagUserAccount FlatMap Exception input: " + value);
                e.printStackTrace();
            }
        }

        private void initMobileAreaMap() throws IOException {
            File mobArea =
                    getRuntimeContext()
                            .getDistributedCache()
                            .getFile(Stage2GlobalSetting.MOBAREAFILENAME);
            mobAreaMap = new HashMap<>();
            List<String> mobAreaLines = FileUtils.readLines(mobArea);
            for (String line : mobAreaLines) {
                String[] splits = line.split(",");
                if (splits.length == 3) {
                    String mobile = splits[0];
                    String province = splits[1];
                    String city = splits[2];
                    /** 维表异常数据处理 */
                    if (province == null || province.equals("")) {
                        province = "null";
                    }

                    if (city == null || city.equals("")) {
                        city = "null";
                    }

                    String mapKey = mobile;
                    String mapVal = province + SPLITWORD + city;
                    mobAreaMap.put(mapKey, mapVal);
                } else {
                    LOG.error("DimPhoneRecordLabel initV1CleanMap error read line: " + line);
                }
            }
        }
    }

    public int getUpdated_at() {
        return updated_at;
    }

    public long getUid() {
        return uid;
    }

    public String getUser_key() {
        return user_key;
    }

    public String getMobile() {
        return mobile;
    }
}
